package spring.cloud.tasks.executor.name_space.executor;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import spring.cloud.tasks.common.node.BlockNodePaths;
import spring.cloud.tasks.common.node.TaskNodePath;
import spring.cloud.tasks.executor.name_space.executor.task.TaskExecutor;
import spring.cloud.tasks.executor.name_space.executor.task.task_listener.task_configuration.TaskConfiguration;
import spring.cloud.tasks.executor.utils.LogUtils;

@Slf4j
@Setter
@Getter
class TaskInitListener implements TreeCacheListener {

    private final TaskInitService taskInitService;

    public TaskInitListener(TaskInitService taskInitService) {
        this.taskInitService = taskInitService;
    }

    @Override
    public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent treeCacheEvent) throws Exception {
        if (treeCacheEvent == null) {
            return;
        }
        ChildData childData = treeCacheEvent.getData();
        if (childData == null) {
            return;
        }
        String path = childData.getPath();
        if (path == null || path.equals(BlockNodePaths.ENABLE_TASKS_NODE_PATH)) {
            return;
        }
        String taskId = path.substring(BlockNodePaths.ENABLE_TASKS_NODE_PATH.length());
        TreeCacheEvent.Type type = treeCacheEvent.getType();
        if (type == null) {
            return;
        }
        if (type.equals(TreeCacheEvent.Type.NODE_ADDED) || type.equals(TreeCacheEvent.Type.NODE_UPDATED)) {
            LogUtils.info(log, taskId, "new job: {} 's jobClass created event received", taskId);
            if (!taskInitService.taskIdSet.contains(taskId)) {
                if (initTaskExecutor(taskId, taskInitService)) {
                    taskInitService.taskIdSet.add(taskId);
                    LogUtils.info(log, taskId, "the job {} initialize successfully", taskId);
                }
            } else {
                LogUtils.warn(log, taskId, "the job {} is unnecessary to initialize, because it's already existing", taskId);
            }
        } else if (type.equals(TreeCacheEvent.Type.NODE_REMOVED)) {
            if (taskInitService.taskIdSet.contains(taskId)) {
                TaskExecutor taskExecutor = ExecutorIdToTaskIdToTaskExecutorMapMap.getTaskExecutor(taskInitService.executorId, taskId);
                if (taskExecutor != null) {
                    taskExecutor.shutdown();
                }
            }
        }

    }

    private boolean initTaskExecutor(String taskId, TaskInitService taskInitService) {
        try {
            LogUtils.info(log, taskId, "start to initialize the new job");
            TaskConfiguration taskConfiguration = new TaskConfiguration(taskInitService.nameSpaceDataManager, taskId);
            if (taskConfiguration.isDeleting()) {
                String executorNodePath = TaskNodePath.getTaskExecutorNode(taskId, taskInitService.executorId);
                taskInitService.nameSpaceDataManager.delete(executorNodePath);
                LogUtils.warn(log, taskId, "the job is on deleting");
                return false;
            }
            TaskExecutor taskExecutor = new TaskExecutor(taskInitService.nameSpaceDataManager, taskInitService, taskConfiguration);

            taskExecutor.init();
            return true;
        } catch (RuntimeException e) {
            LogUtils.warn(log, taskId, "job initialize failed, but will not stop the init process", e);
        } catch (Throwable t) {
            LogUtils.warn(log, taskId, "job initialize failed, but will not stop the init process", t);
        }
        return false;
    }
}
